package com.globalwave.common.scheduling;

import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ScheduledFuture;
import java.util.concurrent.TimeUnit;

import org.apache.commons.lang.StringUtils;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.scheduling.Trigger;
import org.springframework.scheduling.concurrent.ThreadPoolTaskScheduler;
import org.springframework.scheduling.support.PeriodicTrigger;
import org.springframework.util.ErrorHandler;
import org.springframework.web.context.WebApplicationContext;

import com.globalwave.common.exception.BusinessException;
import com.globalwave.common.task.SyncControlTask;
import com.globalwave.common.task.SyncBatchTask;
import com.globalwave.datasync.SyncConfig;
import com.globalwave.system.entity.TaskConfig;
import com.globalwave.system.entity.TaskConfigCriterion;
import com.globalwave.system.service.TaskConfigService;


/**
 * 
 * Job:
 * start all/cancel all exception处理
 * 
 * @author CN084703
 *
 */
public class TaskScheduler {

	private Log log = LogFactory.getLog(this.getClass()) ;
	
	private static final TaskScheduler instance = new TaskScheduler() ;
	
	private ThreadPoolTaskScheduler taskScheduler = new ThreadPoolTaskScheduler() ;
	private WebApplicationContext  appContext ;
	private Map<Long, ScheduledFuture<Object>> scheduledFutures = 
		new Hashtable<Long, ScheduledFuture<Object>>() ;
	
	public static TaskScheduler getInstance(){
		return instance ;
	}
	
	public void initialize(WebApplicationContext  appContext) {

    	if (taskScheduler == null) {
    		taskScheduler = new ThreadPoolTaskScheduler() ;
    	}
    	this.appContext = appContext ;
    	
    	taskScheduler.setThreadNamePrefix("SpringCronJob");
    	taskScheduler.initialize();
    	
		taskScheduler.setErrorHandler(new ErrorHandler() {				
			@Override
			public void handleError(Throwable t) {
				log.error("Error Handler for Task Run:", t) ;
				throw new RuntimeException(t); 
			}
		}) ;
	}
	
	public void shutdown() {
		if (taskScheduler != null) {
			taskScheduler.getScheduledExecutor().shutdown();
			taskScheduler = null ;
		}
	}

	public void scheduleAll() {

		try{
			TaskConfigService taskConfigService = (TaskConfigService)appContext.getBean("taskConfigService") ;
			TaskConfigCriterion criterion = new TaskConfigCriterion() ;
			criterion.setTask_status(TaskConfig.TASK_STATUS_ACTIVE) ;
			
			List<TaskConfig> configs = taskConfigService.queryAll(criterion) ;
			
			for (TaskConfig config:configs) {
				schedule(config) ;
			}
		} catch (Exception e) {
			log.error("schedule all task fail!",e) ;
		}
	}

	public void schedule(Long configId) throws Exception {
		TaskConfigService taskConfigService = (TaskConfigService)appContext.getBean("taskConfigService") ;
		TaskConfigCriterion criterion = new TaskConfigCriterion() ;
		criterion.setTask_status(TaskConfig.TASK_STATUS_ACTIVE) ;
		criterion.setTask_config_id(configId) ;
		
		List<TaskConfig> configs = taskConfigService.queryAll(criterion) ;
		if (configs.size() == 0) {
			// E0101020001', '没有找到任务配置！
			throw new BusinessException("E0101020001")  ;//"not found config!"
		} else {
			schedule(configs.get(0)) ;
		}
		
	}
	
	public void schedule(TaskConfig config) {

		TaskContext context = new TaskContext() ;
		String paramsAsStr = config.getCommand_params() ;
		
		if (!StringUtils.isEmpty(paramsAsStr)) {
			String[] paramsKeyValue = paramsAsStr.split("&") ;
			for (String keyValue:paramsKeyValue) {
				String[] params = keyValue.split("=") ;
			    context.put(params[0], params[1]) ;
			}
		}
		
		try {
			RunnableTask task = null ;
			
			final String commandType = config.getCommand_type() ;
			final String content = config.getCommand_content() ;
			
			if (TaskConfig.COMMAND_TYPE_JAVA.equals(commandType)) {
			    task = (RunnableTask)Class.forName(content).newInstance() ;
			} else if (TaskConfig.COMMAND_TYPE_DATA_SYNC.equals(commandType)) {
				if (SyncConfig.MODE_INCREMENT_INSERT.equals(content) || "IncrementInsertSyncTask".equals(content)) {
					SyncControlTask icTask = new SyncControlTask() ;
					task = icTask ;
					context.put("SYNC_MODE", SyncConfig.MODE_INCREMENT_INSERT) ;
				} else if (SyncConfig.MODE_ONE_BY_ONE.equals(content) || "OneByOneSyncTask".equals(content)) {
					SyncBatchTask oTask = new SyncBatchTask() ;
					task = oTask ;
					context.put("SYNC_MODE", SyncConfig.MODE_ONE_BY_ONE) ;
				} else if (SyncConfig.MODE_MOVE.equals(content) || "MoveSync".equals(content)) {
					SyncControlTask icTask = new SyncControlTask() ;
					task = icTask ;
					context.put("SYNC_MODE", SyncConfig.MODE_MOVE) ;
				}
				
			}
			task.setTaskConfig(config) ;
			//task.setTaskLogService((TaskLogService)appContext.getBean("taskLogService")) ;
			task.setContext(context) ;
			
			Trigger trigger = null ;

			String taskType = config.getTask_type() ;
			
	    	if (TaskConfig.TASK_TYPE_SCHEDULING.equals(taskType)) {
	    		trigger = new CronTrigger(config.getCron_sequence()) ;
	    	} else if (TaskConfig.TASK_TYPE_PERIOD.equals(taskType)) {
	    		trigger = new PeriodicTrigger(config.getTrigger_period(), TimeUnit.SECONDS) ;
	    	} else {
	    		trigger = new OneTimeTrigger(0) ;
	    	}

			schedule(task, trigger) ;
			
		} catch (Exception e) {
			log.error(e, e) ;
		}
	}
	/**
	 * start a task
	 * @param task
	 * @param trigger
	 */
	private void schedule(RunnableTask task, Trigger trigger) {
		Long id = task.getTaskConfig().getTask_config_id() ;
		cancel(id) ;
		scheduledFutures.put(id, taskScheduler.schedule(task, trigger)) ;

	}
	

	public ScheduledFuture<Object> getScheduledFutures(Long configId) {
		return scheduledFutures.get(configId) ;
	}
	
	public boolean isScheduled(Long configId) {
		return scheduledFutures.containsKey(configId) ;
	}


	public void cancelAll() {

		try{/*
			TaskConfigService taskConfigService = (TaskConfigService)appContext.getBean("taskConfigService") ;
			TaskConfigCriterion criterion = new TaskConfigCriterion() ;
			criterion.setTask_status(TaskConfig.TASK_STATUS_ACTIVE) ;
			
			List<TaskConfig> configs = taskConfigService.queryAll(criterion) ;
			
			for (TaskConfig config:configs) {
				cancel(config.getTask_config_id()) ;
			}*/
			for (Long id:scheduledFutures.keySet()) {
				cancel(id) ;
			}
		} catch (Exception e) {
			log.error("schedule all task fail!",e) ;
		}
		
	}
	
	/**
	 * end a task
	 */
	public void cancel(Long configId) {
		ScheduledFuture future = scheduledFutures.get(configId) ;
		if (future != null) {
		    future.cancel(true) ;
		    scheduledFutures.remove(configId) ;
		}
	}
	
	@Override
	protected void finalize() throws Throwable {
		shutdown() ;
		super.finalize();
	}
}
